package ruc.TDEngine.write;

import okhttp3.Response;
import ruc.Consts.constant;
import ruc.utils.IOUtils;
import ruc.TDEngine.TDEngine_Common;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class TDEngine_ConcurrentWrite {
    private String IP, port, dbName, tbName,userName,password;

    TDEngine_ConcurrentWrite(String IP, String port, String dbName, String tbName, String userName, String password) {
        this.IP = IP;
        this.port = port;
        this.dbName = dbName;
        this.tbName = tbName;
        this.userName = userName;
        this.password = password;
    }

    private long splitAndSend(String[] arr, int st, int ed) {
        StringBuilder sql = new StringBuilder(String.format("insert into %s.%s values", dbName, tbName));
        long costTime = 0;
        int cnt = 0;
        for(int i = st; i < ed; i++) {
            cnt++;
            String[] vales = arr[i].split(":");
            String tmp = String.format(" (%s, \"%s\", \"%s\", \"%s\", %s, %s, %s)", vales[0], vales[1], vales[2], vales[3], vales[4], vales[5], vales[6]);
            sql.append(tmp);

            if(cnt > constant.threshold) {
                long st_time = System.nanoTime();
                Response response = TDEngine_Common.send(TDEngine_Common.getRequest(sql.toString(), userName, password, IP, port));
                assert response != null;
                response.close();
                long ed_time = System.nanoTime();

                costTime += (ed_time - st_time) / constant.Ns2Ms;
                cnt = 0;
                sql = new StringBuilder(String.format("insert into %s.%s values", dbName, tbName));
            }
        }

        if(cnt > 0) {
            long st_time = System.nanoTime();
            Response response = TDEngine_Common.send(TDEngine_Common.getRequest(sql.toString(), userName, password, IP, port));
            assert response != null;
            response.close();
            long ed_time = System.nanoTime();

            costTime += (ed_time - st_time) / constant.Ns2Ms;
        }

        return costTime;
    }

    private void MulInsertData(int[] ThreadNum, String[] val) throws InterruptedException {
        for(int i = 0; i < ThreadNum.length; i++) {
            TDEngine_Common.dropAndCreate(constant.IP, constant.TDEngine_port, constant.dbName, constant.Concurrent_tbName, constant.userName, constant.passWord);

            CountDownLatch CDL = new CountDownLatch(ThreadNum[i]);
            AtomicLong costTime = new AtomicLong(0);

            for(int k = 0; k < ThreadNum[i]; k++) {
                int K = k;
                int I = i;
                new Thread(()->{
                    int st = K * (val.length / ThreadNum[I]);
                    int ed = K == ThreadNum[I] - 1 ? val.length : (K + 1) * (val.length / ThreadNum[I]);
                    costTime.addAndGet(splitAndSend(val, st, ed));

                    CDL.countDown();
                }).start();
            }

            CDL.await();
            System.out.printf("[TSBench-D] [TDEngine_ConcurrentWrite] >>>>>> threadNum %d, insert time %d ms >>>>>> \n", ThreadNum[i], costTime.longValue());
            IOUtils.WriteData(constant.GetWriteFilePath("TDEngine", "ConcurrentWrite"),
                    new String[]{String.valueOf(ThreadNum[i]), String.valueOf(costTime.longValue())});
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TDEngine_ConcurrentWrite root = new TDEngine_ConcurrentWrite(constant.IP, constant.TDEngine_port, constant.dbName, constant.Concurrent_tbName, constant.userName, constant.passWord);
        root.MulInsertData(constant.ConWrite_Thread, IOUtils.GetData());
    }
}
